#pragma once
#include <ev.h>
#include <amqpcpp.h>
#include <amqpcpp/libev.h>
#include <openssl/ssl.h>
#include <openssl/opensslv.h>
#include <iostream>
#include <functional>
#include "logger.h"

class MQClient
{
    using MessageCallback = std::function<void(const char *, size_t)>;
    using ptr = std::shared_ptr<MQClient>;

public:
    MQClient(const std::string &user, const std::string &passwd, const std::string &host)
    {
        // 实例化底层网络通信框架的I/O事件监控句柄
        _loop = EV_DEFAULT;
        // 实例化libEvHandler句柄 --- 将AMQP框架与事件监控关联起来
        _handler = std::make_unique<AMQP::LibEvHandler>(_loop);
        // amqp://root:123456@127.0.0.1:5672/
        std::string url = "amqp://" + user + ":" + passwd + "@" + host + "/";
        // 实例化连接对象
        AMQP::Address address(url);
        //
        _connection = std::make_unique<AMQP::TcpConnection>(_handler.get(), address);
        // 实例化通信信道
        _channel = std::make_unique<AMQP::TcpChannel>(_connection.get());
        // 启动底层网络通信框架--开启I/O
        _loop_thread = std::thread([this]()
                                   { ev_run(_loop, 0); });
    }
    ~MQClient()
    {
        // 异步线程调用
        ev_async_init(&_async_watcher, watcher_callback); // 初始化
        ev_async_start(_loop, &_async_watcher);           // 开始
        ev_async_send(_loop, &_async_watcher);            // 结束
        _loop_thread.join();                              // 等待
        _loop = nullptr;
    }
    void declareComponents(const std::string &exchange,
                           const std::string &queue,
                           const std::string &routing_key = "routing_key",
                           AMQP::ExchangeType echange_type = AMQP::ExchangeType::direct)
    {
        // 声明交换机
        _channel->declareExchange(exchange, echange_type)
            .onError([](const char *message)
                     {
                    LOG_ERROR("声明交换机失败：{}", message);
                    exit(0); })
            .onSuccess([exchange]()
                       { LOG_ERROR("{} 交换机创建成功！", exchange); });
        // 声明消息队列
        _channel->declareQueue(queue)
            .onError([](const char *message)
                     {
                    LOG_ERROR("声明消息队列失败：{}", message);
                    exit(0); })
            .onSuccess([queue]()
                       { LOG_ERROR("{} 交换机创建成功！", queue); });
        // 针对交换机和队列进行绑定
        _channel->bindQueue(exchange, queue, routing_key)
            .onError([exchange, queue](const char *message)
                     {
                    LOG_ERROR("{} - {} 绑定失败：", exchange, queue);
                    exit(0); })
            .onSuccess([exchange, queue, routing_key]()
                       { LOG_ERROR("{} - {} - {} 绑定成功！", exchange, queue, routing_key); });
    }
    // 向交换机发送消息
    bool pubish(const std::string &exchange,
                const std::string &msg,
                const std::string &routing_key = "routing_key")
    {
        LOG_DEBUG("向交换机 {}-{} 发布消息！", exchange, routing_key);
        bool ret = _channel->publish(exchange, routing_key, msg);
        if (!ret)
        {
            LOG_ERROR("{} 发布消息失败：", exchange);
            return false;
        }
        return true;
    }
    // 查询
    bool consume(const std::string &queue, const MessageCallback &cb)
    {
        LOG_DEBUG("开始订阅 {} 队列消息！", queue);
        // 开始查询返回值 DeferredConsumer
        _channel->consume(queue, "consume-tag")
            .onReceived([this, cb](const AMQP::Message &message,
                                   uint64_t deliveryTag,
                                   bool redelivered)
                        {
                    cb(message.body(), message.bodySize());
                    _channel->ack(deliveryTag); })
            .onError([queue](const char *message)
                     {
                    LOG_ERROR("订阅 {} 队列消息失败: {}", queue, message);
                    exit(0); });
    }

private:
    // 监视回调函数
    static void watcher_callback(struct ev_loop *loop, ev_async *watcher, int32_t revents)
    {
        ev_break(loop, EVBREAK_ALL);
    }

private:
    struct ev_async _async_watcher;                   // 异步线程
    struct ev_loop *_loop;                            // 监控句柄
    std::unique_ptr<AMQP::LibEvHandler> _handler;     // 文件描述符处理
    std::unique_ptr<AMQP::TcpConnection> _connection; // tcp建立链接
    std::unique_ptr<AMQP::TcpChannel> _channel;       // 信道通道
    std::thread _loop_thread;                         // 线程
};
